/*
 * Su-yun Information Technology, Co.,Ltd CONFIDENTIAL
 *
 * (C) Copyright Su-yun Information Technology, Co.,Ltd. 2021 All Rights Reserved.
 *
 * NOTICE:  All information contained herein is, and remains the property of
 * Su-yun Information Technology, Co.,Ltd. The intellectual property herein
 * are protected by trade secret, patent law or copyright law. Dissemination
 * of this information or reproduction of this material is strictly forbidden
 * unless prior written permission is obtained from Su-yun Information Technology.
 *
 * https://suyunplat.com/
 *
 */
package com.xtiim.battery.discharge.protocal.manager;

import cn.hutool.core.util.NumberUtil;
import cn.hutool.core.util.StrUtil;
import com.google.common.collect.Maps;
import com.xtiim.battery.discharge.common.Configuration;
import com.xtiim.battery.discharge.common.Constants;
import com.xtiim.battery.discharge.common.concurrent.pool.GlobalExecutor;
import com.xtiim.battery.discharge.event.*;
import com.xtiim.battery.discharge.protocal.ResponseReader;
import com.xtiim.battery.discharge.protocal.command.BatteryCommand;
import com.xtiim.battery.discharge.protocal.command.CommandType;
import com.xtiim.battery.discharge.protocal.event.*;
import com.xtiim.battery.discharge.util.ByteUtil;
import com.xtiim.battery.discharge.util.SerialPortUtil;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import purejavacomm.SerialPort;

import java.util.Arrays;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * Manage and handle serial port communication
 */
@Slf4j
public class CommunicationManager implements ApplicationListener<ResponseReceivedEvent> {

    private static final int SEND_CMD_RETRY_TIMES = 3;

    private static CommunicationManager instance = new CommunicationManager();

    private BlockingQueue<BatteryCommand> commandTaskQueue = new LinkedBlockingQueue<>(1024 * 1024);

    private Map<String, PortState> portStateMap = Maps.newConcurrentMap();

    private EventPublisher eventPublisher = DefaultEventPublisher.getInstance();

    private GlobalExecutor globalExecutor = GlobalExecutor.getInstance();

    private EventMulticaster eventMulticaster = DefaultEventMulticaster.getInstance();

    private SendCommandHandler sendCommandHandler = new SendCommandHandler();

    private ResponseTimeoutChecker timeoutChecker = new ResponseTimeoutChecker();

    public BufferHandler bufferHandler = new BufferHandler();

    private static int responseTimeout = 1000;

    public static final byte EB = new Integer(0xEB).byteValue();
    public static final byte ED = new Integer(0xED).byteValue();

    private CommunicationManager() {
    }

    public static CommunicationManager getInstance() {
        return instance;
    }


    /**
     * Initialize background tasks and event listener
     */
    public void init() {
        globalExecutor.submitSendCommand(sendCommandHandler);
        globalExecutor.scheduleTimeoutCheck(timeoutChecker);

        eventMulticaster.register(ResponseReceivedEvent.class, this);

        String respTimeoutParam = Configuration.get(Constants.RESPONSE_TIMEOUT);
        if (StrUtil.isNotBlank(respTimeoutParam) && NumberUtil.isInteger(respTimeoutParam)) {
            responseTimeout = Integer.valueOf(respTimeoutParam);
        }
    }

    /**
     * Add command into queue
     *
     * @param command
     */
    public void addCommand(BatteryCommand command) {
        if (command == null || command.getCmd() == null || command.getCommandType() == null) {
            throw new IllegalArgumentException("Invalid command param.");
        }
        if (command.getPort() == null) {
            throw new IllegalArgumentException("Serial port is null.");
        }

        boolean success = commandTaskQueue.offer(command);
        if (!success) {
            log.error("Add command to queue failed. queueSize: {}. port: {}. commandType: {}",
                    commandTaskQueue.size(), command.getPort().getName(), command.getCommandType().name());
        }
    }

    @Override
    public void onApplicationEvent(ResponseReceivedEvent event) {
        byte[] respBytes = event.getRespBytes();
        if (respBytes.length == 1 && respBytes[0] == ResponseReader.COMM_RESPONSE) {
            eventPublisher.publishEvent(new CommandRepliedEvent(this, event.getPortName()));
            return;
        }

        //if the response is not a normal response, ignore it.
        if (respBytes.length > 1) {
            if (respBytes[0] != EB || respBytes[respBytes.length - 2] != ED) {
                log.warn("Illegal protocol bytes: ==========> ");
                return;
            }
        }

        PortState portState = portStateMap.computeIfAbsent(event.getPortName(), e -> new PortState());
        portState.setStatus(ResponseStatus.NORMAL);
        try {
            bufferHandler.processBuffer(event.getPortName(), portState, event.getRespBytes());
        } catch (Exception e) {
            log.error("Error occurs in buffer handling. error: {}", ExceptionUtils.getStackTrace(e));
        }
    }

    /**
     * Packet bytes may be split and sent by multiples times.
     * So it is necessary to buffer a complete packet before
     * handle it.
     */
    public class BufferHandler {

        /**
         * Process packet bytes buffer.
         *
         * @param portState
         * @param receivedBytes
         */
        public void processBuffer(String port, PortState portState, byte[] receivedBytes) {
            PacketBuf packetBuf = portState.getPacketBuf();
            int current = 0;
            while (current < receivedBytes.length) {
                if (packetBuf.bufSize() > 0) {
                    if (packetBuf.endByte() == ED) {
                        packetBuf.append(receivedBytes, current, 1);
                        current++;
                        sendEvent(port, packetBuf.transfer());
                    } else {
                        int start = current;
                        while (current < receivedBytes.length - 1 && receivedBytes[current] != ED) {
                            current++;
                        }
                        packetBuf.append(receivedBytes, start, current - start + 1);
                        current++;
                    }
                } else {
                    if (receivedBytes[current] != EB) {
                        throw new IllegalStateException("Illegal protocol bytes.");
                    } else {
                        int start = current;
                        while (current < receivedBytes.length - 1 && receivedBytes[current] != ED) {
                            current++;
                        }
                        packetBuf.append(receivedBytes, start, current - start + 1);
                        current++;
                    }
                }
            }
        }

        /**
         * Send event
         *
         * @param port
         * @param bytes
         */
        private void sendEvent(String port, byte[] bytes) {
            CommunicationManager.this.eventPublisher.publishEvent(new PacketReadyEvent(this, port, bytes));
        }
    }

    /**
     * Send command handler
     */
    private class SendCommandHandler implements Runnable {

        @Override
        public void run() {
            for (; ; ) {
                try {
                    BatteryCommand command = commandTaskQueue.take();
                    handle(command);
                } catch (Throwable e) {
                    log.error("Exception occurs when take and handle command task. {}", ExceptionUtils.getStackTrace(e));
                }
            }
        }

        /**
         * Handle command sending.
         * <p>
         * After either command or collect request is sent,
         * port is in Waiting status. But still allow sending
         * next command and next because the communication may
         * recover at anytime. The last send time become
         * more distant past until the timeout check task
         * decide the port is timeout, or some response get back
         * to reset to Ready status.
         *
         * @param command
         */
        private void handle(BatteryCommand command) {
            String host = command.getHost();
            SerialPort port = command.getPort();
            CommandType commandType = command.getCommandType();
            log.info("host: ====> {}, commandType: ====> {}", host, commandType);
            CommandMatcher commandMatcher = CommandMatcher.getInstance(host);
            if (!commandMatcher.match(host, commandType)) {
                return;
            }

            byte[] cmd = command.getCmd();

            boolean success = false;
            for (int i = 0; i < SEND_CMD_RETRY_TIMES; i++) {
                success = SerialPortUtil.sendToPort(port, cmd);
                if (success) {
                    break;
                }
                log.error("[Send command failed] port: {}, cmd: {}, times: {}", port.getName(),
                        ByteUtil.byte2hex(cmd), i + 1);
                try {
                    TimeUnit.MICROSECONDS.sleep(10);
                } catch (InterruptedException e) {
                    log.warn("Sleep is interrupted while retry sending cmd.");
                }
            }
            if (success) {
                log.info("[Send command success] port: {}, cmd: {}", port.getName(), ByteUtil.byte2hex(cmd));
                PortState portState = portStateMap.computeIfAbsent(command.getPort().getName(), e -> new PortState());
                portState.setLastSendTime(new Date());
                portState.setStatus(ResponseStatus.WAITING);

                //count send command num
                HostManager.getInstance().countSendCommandNum(host);

            } else {
                eventPublisher.publishEvent(new CommandFailureEvent(this, port.getName()));
            }
        }
    }

    /**
     * Response timeout checker.
     */
    private class ResponseTimeoutChecker implements Runnable {

        private EventPublisher eventPublisher = DefaultEventPublisher.getInstance();

        @Override
        public void run() {
            for (String port : portStateMap.keySet()) {
                try {
                    PortState portState = portStateMap.get(port);
                    if (portState != null && ResponseStatus.WAITING.equals(portState.getStatus())) {
                        Long elapsed = Long.valueOf(System.currentTimeMillis() - portState.getLastSendTime().getTime()).longValue();
                        if (elapsed > responseTimeout) {
                            eventPublisher.publishEvent(new CommandTimeoutEvent(this, port));
                            log.debug("Command response timeout. port: {}", port);
                        }
                    }
                } catch (Exception e) {
                    log.error("Exception occurs when checking port timeout. {}", ExceptionUtils.getStackTrace(e));
                }
            }
        }
    }

    /**
     * Port state which stores last send time
     * and current status.
     */
    @Data
    public static class PortState {
        private Date lastSendTime;
        private ResponseStatus status;
        private PacketBuf packetBuf;

        public PortState() {
            this.lastSendTime = null;
            this.status = ResponseStatus.NORMAL;
            this.packetBuf = new PacketBuf();
        }
    }

    /**
     * Encapsulate packet data receive buffer
     */
    @Data
    @AllArgsConstructor
    private static class PacketBuf {
        private byte[] buf;
        private int bufLen;

        public PacketBuf() {
            // protocol is one-request-one-response model.
            // so at most one packet fly in air.
            // so maximum packet len buffer is enough.
            this.buf = new byte[256];
            this.bufLen = 0;
        }

        /**
         * Append new bytes at buffer pointer
         *
         * @param bytes
         * @param start
         * @param len
         */
        public void append(byte[] bytes, int start, int len) {
            if (len > freeSize()) {
                throw new IllegalArgumentException("Given bytes beyond the free buffer size.");
            }
            System.arraycopy(bytes, start, buf, bufLen, len);
            bufLen += len;
        }

        /**
         * Transfer effective buffer bytes to invoker
         * and reset buffer pointer to zero location.
         *
         * @return
         */
        public byte[] transfer() {
            byte[] bytes = new byte[bufLen];
            System.arraycopy(buf, 0, bytes, 0, bufLen);

            Arrays.fill(buf, 0, buf.length, (byte) 0);
            bufLen = 0;

            return bytes;
        }

        /**
         * Answer the free size
         *
         * @return
         */
        public int freeSize() {
            return buf.length - bufLen;
        }

        /**
         * Answer the buffer size
         *
         * @return
         */
        public int bufSize() {
            return bufLen;
        }

        /**
         * Answer end byte
         *
         * @return
         */
        public byte endByte() {
            if (bufLen == 0) {
                throw new IllegalArgumentException("There is no end byte since buffer is empty.");
            }
            return getByte(bufLen - 1);
        }

        /**
         * Answer byte at the given index
         *
         * @param index
         * @return
         */
        private byte getByte(int index) {
            if (index < 0 || index > bufLen - 1) {
                throw new IllegalArgumentException("Index out of buffer boundary.");
            }
            return buf[index];
        }
    }

    /**
     * Response status on port
     */
    private enum ResponseStatus {
        NORMAL, WAITING;
    }
}
